package cn.py.boot_lua.asyn;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.function.Consumer;

/***
  * @Message 异步任务引擎执行器
 */

@Slf4j
public class AsyncQueueTaskEngine<T>  {

    private final BlockingQueue<T> taskQueue;
    private final Executor executor;
    private final Semaphore semaphore; // 并发控制
    private final Consumer<T> taskHandler;
    private volatile boolean running = true;
    private final Thread consumerThread;

    public AsyncQueueTaskEngine(int queueCapacity, Executor executor, int maxConcurrency, Consumer<T> taskHandler) {
        this.taskQueue = new LinkedBlockingQueue<>(queueCapacity);
        this.executor = executor;
        this.semaphore = maxConcurrency > 0 ? new Semaphore(maxConcurrency) : null;
        this.taskHandler = taskHandler;

        this.consumerThread = new Thread(this::consumeLoop, "AsyncQueueTaskEngine-Consumer");
        this.consumerThread.setDaemon(true);
        this.consumerThread.start();
        log.info("AsyncQueueTaskEngine started with queueCapacity={}, maxConcurrency={}", queueCapacity, maxConcurrency);
    }

    public void submitTask(T task) throws InterruptedException {
        log.info("提交任务到队列: {}", task);
        taskQueue.put(task);
        log.info("任务成功入队: {}", task);
    }

    public void shutdown() {
        running = false;
        consumerThread.interrupt();
        log.info("AsyncQueueTaskEngine已关闭");
    }

    private void consumeLoop() {
        while (running) {
            try {
                log.debug("等待获取新任务...");
                T task = taskQueue.take();
                log.info("获取到任务，准备处理: {}", task);

                Runnable runnable = () -> {
                    try {
                        log.info("开始执行任务: {}", task);
                        taskHandler.accept(task);
                        log.info("任务执行完成: {}", task);
                    } catch (Exception e) {
                        log.error("任务执行异常: {}", task, e);
                    } finally {
                        if (semaphore != null) {
                            semaphore.release();
                            log.debug("任务完成，释放信号量");
                        }
                    }
                };

                if (semaphore != null) {
                    log.debug("尝试获取信号量，当前可用许可数: {}", semaphore.availablePermits());
                    semaphore.acquire();
                    log.debug("信号量获取成功，继续执行任务");
                }

                executor.execute(runnable);

            } catch (InterruptedException e) {
                if (!running) {
                    log.info("消费者线程被中断，退出消费循环");
                    Thread.currentThread().interrupt();
                    break;
                }
            } catch (Exception e) {
                log.error("任务消费循环异常", e);
            }
        }
        log.info("消费者线程退出");
    }

}
